1：什么是 MCP / MCP Server

The Model Context Protocol (MCP) is an open protocol that enables seamless integration between LLM applications and external data sources and tools. Whether you're building an AI-powered IDE, enhancing a chat interface, or creating custom AI workflows, MCP provides a standardized way to connect LLMs with the context they need.

上面是 MCP 的官方介绍，用通俗的语言来说 MCP 就是连接 AI 与万物的桥梁，也就是说 AI 可以通过这个桥梁去操作理论上的任何东西，多余的我们就不过多赘述了，我们这次的目的就是为了去开发一个 MCP Server，来看看这个过程是如何做的

2：基于 stdio 的 MCP

MCP 官网提供了 Python 的 SDK Python-SDK，我们可以轻松的通过这个 SDK 去编写一个非常简单的 MCP，只不过这里我们使用的是基于 stdio 的方式来实现，后面会有其他两种方式的实现的案例

1：使用 uv 工具初始化项目

2：安装 MCP-SDK

3：编写 MCP Server

如果你是在别的 MCP Client 上使用的话，你可以将如下 JSON 直接粘贴到添加 MCP 服务器中

那么，这个就是基础的 MCP Server 使用本地的 stdio 的方式调用它

3：基于 SSE 的 MCP

其实修改为 SSE 协议非常的简单，我们来看代码

这样修改完，我们需要本地将其启动，因为这就成了一个远程调用的方式了，我们就需要它监听端口了

然后我们同样的去 MCP 客户端去添加这个 MCP 服务器

可以看到，我们没用使用 stdio 的方式依旧调用到了 add 的方法，这里使用的就是 sse 协议

方法	说明
stdio	一般需要客户端将 mcp server（也就是 python 程序）下载到客户端所在的主机上然后才可以配置调用它
sse	一般只需要将 mcp server（也就是 python 程序）部署到一个可以被访问的服务器或者主机上即可，只需要对外提供 IP+PORT 即可被调用

3：基于 Streamable http 的 MCP

它和 SSE 其实是非常的相似的，我们只需要将 sse 修改为 stremable-http 就可以使用它了

Streamable HTTP 的优势

优势	说明
更简单	Streamable HTTP 只需要使用标准的HTTP 请求，无需建立长连接，客户端的实现更简单，代码量更少，维护成本更低。
更高的性能	Streamable HTTP 避免了长连接的开销，响应时间更短，稳定性更好。
更好的无状态服务器支持	Streamable HTTP 可以更好地支持无状态服务器，因为无需维持长连接，减轻了服务器的压力。
更广泛的兼容性	Streamable HTTP 是一种纯HTTP 实现，可以与现有的基础设施很好地集成。

SSE 的优势

优势	说明
双向通信	SSE 允许服务器和客户端进行双向通信，客户端可以向服务器发送事件，服务器也可以向客户端推送事件。
事件驱动	SSE 采用事件驱动的方式，服务器可以根据需要向客户端推送不同类型的事件。
广泛的应用场景	SSE 广泛应用于实时更新数据，例如新闻更新、聊天信息或股票价格等。

Streamable http 对比 SSE

特点	Streamable HTTP	SSE
连接方式	标准 HTTP 请求，不维持长连接	建立长连接，持续接收事件
实现复杂度	客户端实现简单，代码量少	客户端实现复杂，需要处理长连接
性能	响应时间短，稳定性好	可能受长连接影响，性能略低
应用场景	流式数据传输，例如大模型推理结果	实时更新数据，例如聊天信息、新闻等

总结：Streamable HTTP 在性能和实现复杂度方面有优势，更适合需要流式数据传输的场景。SSE 适合实时更新数据，尤其是在需要双向通信和事件驱动的应用场景。在选择Streamable HTTP 还是SSE 时，需要根据实际应用场景和需求来权衡其优缺点。例如，对于需要处理大量数据流的场景，Streamable HTTP 可能是更好的选择，而对于需要实时推送消息和双向通信的场景，SSE 可能更合适。

4：编写简单的 MCP Server on K8S

# 导入 FastMCP 类 from mcp.server.fastmcp import FastMCP from kubernetes import client, config # 创建 MCP Server 实例 mcp = FastMCP("MCP-Server") config.load_kube_config() @mcp.tool() def getPods(namespace: str) -> list: """获取 k8s 中指定的命名空间下的所有的 Pod""" v1 = client.CoreV1Api() # 声明一个 CoreV1Api 实例 return v1.list_namespaced_pod(namespace).items # 获取指定 Namespace 中的所有 Pod # 它可以让 main.py 成为一个独立可运行的程序 if __name__ == "__main__": # 启动 MCP Server mcp.run(transport="streamable-http")